package com.zyx.sparkdemo.mllib.featurescaler

import org.apache.spark.SparkConf
import org.apache.spark.ml.feature.Normalizer
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.SparkSession

/**
 * @author Yaxi.Zhang
 * @since 2021/8/27 16:18
 *        reference: https://blog.csdn.net/neilron/article/details/75329973
 */
object NormalizerDemo {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("NormalizerDemo")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    val dataFrame = spark.createDataFrame(Seq(
      (0, Vectors.dense(1.0, 0.5, -1.0)),
      (1, Vectors.dense(2.0, 1.0, 1.0)),
      (2, Vectors.dense(4.0, 10.0, 2.0))
    )).toDF("id", "features")

    // Normalizer的作用范围是 每一行, 使每一个行向量的范数变换为一个单位范数
    // 正则化每个向量到1阶范数
    val normalizer = new Normalizer()
      .setInputCol("features")
      .setOutputCol("normFeatures")
      .setP(1.0)

    println("Normalized using L^1 norm")
    val l1NormData = normalizer.transform(dataFrame)

    l1NormData.show
    /*
        +---+--------------+------------------+
        | id|      features|      normFeatures|
        +---+--------------+------------------+
        |  0|[1.0,0.5,-1.0]|    [0.4,0.2,-0.4]|
        |  1| [2.0,1.0,1.0]|   [0.5,0.25,0.25]|
        |  2|[4.0,10.0,2.0]|[0.25,0.625,0.125]|
        +---+--------------+------------------+
     */


    // 正则化每个向量到无穷阶范数
    println(Double.PositiveInfinity)

    val lInfNormData = normalizer.transform(dataFrame, normalizer.p -> Double.PositiveInfinity)

    println("Normalized using L^inf norm")
    lInfNormData.show()

    spark.close()
  }
}
